package ca.tatham.senssim.pddd;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Set;

import ca.tatham.senssim.Packet;
import ca.tatham.senssim.SensorNode;
import ca.tatham.senssim.SensorType;
import ca.tatham.senssim.Simulation;
import ca.tatham.senssim.diffusion.AbstractDiffusionNode;
import ca.tatham.senssim.diffusion.DataMessage;
import ca.tatham.senssim.diffusion.InterestMessage;
import ca.tatham.senssim.diffusion.InterestMessage.Interest;
import ca.tatham.senssim.util.Address;

public class SimpleNodePddd extends AbstractDiffusionNode implements SensorNode
{
  public static final int MAX_HEARTBEAT_TIMEOUT = 30000;

  private static final int DELTA = 10;

  private final Set<SensorType> m_sensors = new HashSet<SensorType>();
  private final Map<Address, SinkInfo> m_sinkInfo = new HashMap<Address, SinkInfo>();
  private final Map<Address, Collection<ExpectedAck>> m_expectedAcks = new HashMap<Address, Collection<ExpectedAck>>();

  private int m_sourcePackets;
  private long m_acksSent;

  protected SimpleNodePddd(Address address, int power)
  {
    super(address, power);
  }

  @Override
  public boolean isSource()
  {
    return true;
  }

  @Override
  public boolean isSink()
  {
    return false;
  }

  public void addSensorType(SensorType type)
  {
    m_sensors.add(type);
  }

  public long getNumAcksSent()
  {
    return m_acksSent;
  }

  public void tick()
  {
    for (SinkInfo info : m_sinkInfo.values())
    {
      for (SensorType sensor : m_sensors)
      {
        if (info.shouldSendData(sensor))
        {
          DataMessage basePacket = new DataMessage(info.getSinkAddress(),
              getAddress(), null, sensor);
          m_sourcePackets++;
          for (Address hop : info.getNextHops())
          {
            DataMessage packet = new DataMessage(basePacket, getAddress(), hop);
            // log("Sending " + packet);
            sendPacket(packet);
          }
        }
      }

      if (info.checkExpired())
      {
        linkBreakageDetected(info);
      }
    }
    for (Collection<ExpectedAck> expectedAcks : m_expectedAcks.values())
    {
      for (ExpectedAck expectedAck : expectedAcks)
      {
        if (expectedAck.tick())
        {
          log("Line Breakage Detected: Lost packet #" + expectedAck.sequence
              + " from " + expectedAck.destination);
          linkBreakageDetected(getSinkInfo(expectedAck.sink));
        }
      }
    }
  }

  /**
   * See Section 3.6 of Data Dissemination in Wireless Sensor Networks
   * 
   * @param info
   */
  private void linkBreakageDetected(SinkInfo info)
  {
    if (!info.haveParent())
    {
      info.lostAllLowerLambdaNeighbors();
      sendPacket(new UpdatePacket(getAddress(), info.getSinkAddress(), info
          .getLevel()));
    }
  }

  public char getCloseSymbol()
  {
    return '>';
  }

  public char getOpenSymbol()
  {
    return '<';
  }

  @Override
  protected void processDataMessage(DataMessage packet)
  {
    SensorType sensor = packet.getType();
    SinkInfo info = getSinkInfo(packet.getSinkAddress());

    if (Simulation.useAcks())
    {
      // log("Acking #" + packet.getSequence() + " from "
      // + packet.getOriginalSource());
      m_acksSent++;
      sendPacket(new AcknowledgementMessage(getAddress(), packet));
    }

    if (info.shouldSendData(sensor))
    {
      for (Address hop : info.getNextHops())
      {
        DataMessage newPacket = new DataMessage(packet, getAddress(), hop);
        // log("Forwarding " + newPacket);
        sendPacket(newPacket);
      }
    }

  }

  @Override
  public void processPacket(Packet packet)
  {
    super.processPacket(packet);
    if (packet instanceof AcknowledgementMessage)
    {
      processAck(packet);
    }
    else if (packet instanceof HeartbeatPacket)
    {
      processHeartbeat(packet);
    }
  }

  private void processHeartbeat(Packet packet)
  {
    getSinkInfo(packet.getSource()).setTimeout(MAX_HEARTBEAT_TIMEOUT);
  }

  private void processAck(Packet packet)
  {
    AcknowledgementMessage ack = (AcknowledgementMessage) packet;
    for (Iterator<ExpectedAck> it = m_expectedAcks.get(packet.getSource())
        .iterator(); it.hasNext();)
    {
      ExpectedAck expectedAck = it.next();
      if (expectedAck.sequence == ack.getDataSequence()
          && expectedAck.destination.equals(ack.getSource()))
      {
        // log(packet.toString());
        it.remove();
        break;
      }
    }
  }

  @Override
  public boolean sendPacket(Packet packet)
  {
    if (super.sendPacket(packet))
    {
      if (packet instanceof DataMessage
          && !packet.getDestination().equals(packet.getSinkAddress()))
      {
        addExpectedAck(packet);
      }
      return true;
    }
    return false;
  }

  private void addExpectedAck(Packet packet)
  {
    if (Simulation.useAcks())
    {
      Address destination = packet.getDestination();
      Collection<ExpectedAck> acks = m_expectedAcks.get(destination);
      if (acks == null)
      {
        acks = new LinkedList<ExpectedAck>();
        m_expectedAcks.put(destination, acks);
      }
      acks.add(new ExpectedAck(packet.getSequence(), destination, packet
          .getSinkAddress()));
    }
  }

  /**
   * See Figure 2 of "Data Dissemination in Wireless Sensor Networks"
   */
  @Override
  protected void processInterestMessage(InterestMessage packet)
  {
    Address sinkId = packet.getSinkAddress();
    SinkInfo sinkInfo = getSinkInfo(sinkId);
    sinkInfo.addInterest(packet.getInterest());
    sinkInfo.updateNeighborLevel(packet);
    if (packet.getSource().equals(sinkId))
    {
      sinkInfo.setTimeout(MAX_HEARTBEAT_TIMEOUT);
    }
    if (packet.getSequence() > sinkInfo.getLastSequenceReceived())
    {
      sinkInfo.setLastSequenceReceived(packet.getSequence());
      sinkInfo.updateLevel(packet);
      forwardInterestMessage(packet, sinkInfo);
    }
    else if (packet.getSequence() == sinkInfo.m_lastSequenceReceived)
    {
      if (sinkInfo.getLevel().getLambda() - packet.getLevel().getLambda() > DELTA)
      {
        sinkInfo.updateLevel(packet);
        forwardInterestMessage(packet, sinkInfo);
      }
    }
  }

  private void forwardInterestMessage(InterestMessage packet, SinkInfo sinkInfo)
  {
    InterestMessage newMessage = new InterestMessage(packet, getAddress(),
        packet.getDestination());
    newMessage.setLevel(sinkInfo.getLevel());
    sendPacket(newMessage);
  }

  private SinkInfo getSinkInfo(Address sinkId)
  {
    SinkInfo info = m_sinkInfo.get(sinkId);
    if (info == null)
    {
      info = new SinkInfo(sinkId);
      m_sinkInfo.put(sinkId, info);
    }
    return info;
  }

  static class InterestMonitor
  {
    private final Interest interest;
    private final long m_begin;
    private long m_lastSend = 0;

    InterestMonitor(Interest aInterest)
    {
      interest = aInterest;
      m_begin = Simulation.getTime();
    }

    public boolean shouldSendData()
    {
      if (!isExpired()
          && (m_lastSend == 0 || Simulation.getTime() >= interest.interval
              + m_lastSend))
      {
        m_lastSend = Simulation.getTime();
        return true;
      }
      return false;
    }

    public boolean isExpired()
    {
      return Simulation.getTime() > m_begin + interest.duration;
    }
  }

  static class SinkInfo
  {
    private final Timer m_timer = new Timer();
    private long m_lastSequenceReceived = Long.MIN_VALUE;
    private final Level m_level = new Level();
    private final Map<Address, Level> m_neighborLevel = new HashMap<Address, Level>();
    private final Address m_sinkAddress;

    private final Map<SensorType, InterestMonitor> m_interests = new HashMap<SensorType, InterestMonitor>();

    SinkInfo(Address sinkAddress)
    {
      if (sinkAddress == null)
      {
        throw new IllegalArgumentException("Cannot have null sink address");
      }
      m_sinkAddress = sinkAddress;
    }

    public Collection<Address> getNextHops()
    {
      return m_level.getParents();
    }

    void addInterest(Interest interest)
    {
      m_interests.put(interest.type, new InterestMonitor(interest));
    }

    boolean shouldSendData(SensorType type)
    {
      InterestMonitor interestMonitor = m_interests.get(type);
      if (interestMonitor == null)
      {
        return false;
      }
      if (interestMonitor.isExpired())
      {
        m_interests.remove(type);
      }
      return interestMonitor.shouldSendData();
    }

    Address getSinkAddress()
    {
      return m_sinkAddress;
    }

    void updateLevel(InterestMessage packet)
    {
      m_level.setLambda(packet.getLevel().getLambda() + DELTA);
      m_level.updateSets(packet.getLevel(), packet.getSource());
    }

    public void setTimeout(int timeout)
    {
      m_timer.setTimeout(timeout);
    }

    public boolean checkExpired()
    {
      return m_timer.tick();
    }

    public void setLastSequenceReceived(long sequence)
    {
      m_lastSequenceReceived = sequence;
    }

    public long getLastSequenceReceived()
    {
      return m_lastSequenceReceived;
    }

    /**
     * See Figure 4 of "Data Dissemination in Wireless Sensor Networks"
     */
    public void lostAllLowerLambdaNeighbors()
    {
      int minSiblingLambda = getMinSiblingLambda();
      if (haveSibling() && haveChild())
      {
        getLevel().setLambda(
            (int) Math.floor((getLevel().getLambda() + minSiblingLambda) / 2));
      }
      else
      {
        getLevel().setLambda(minSiblingLambda);
      }
      // TODO updateLevel(packet);
    }

    public int getMinSiblingLambda()
    {
      int result = Integer.MAX_VALUE;
      for (Address key : m_neighborLevel.keySet())
      {
        Level level = m_neighborLevel.get(key);
        if (level.getLambda() < result)
        {
          result = level.getLambda();
        }
      }
      return result;
    }

    private void updateNeighborLevel(InterestMessage packet)
    {
      m_neighborLevel.put(packet.getSource(), packet.getLevel());
    }

    public Level getNeighborLevel(Address node)
    {
      return m_neighborLevel.get(node);
    }

    public boolean haveParent()
    {
      return m_level.haveParent();
    }

    public boolean haveChild()
    {
      return m_level.haveChild();
    }

    public boolean haveSibling()
    {
      return m_level.haveSibling();
    }

    Level getLevel()
    {
      return m_level;
    }

    @Override
    public String toString()
    {
      return "Sink Info for " + getSinkAddress() + " Level: " + getLevel();
    }
  }

  public int getSourcePackets()
  {
    return m_sourcePackets;
  }

  static class Timer
  {
    private int m_timeout;

    Timer()
    {
      this(MAX_HEARTBEAT_TIMEOUT);
    }

    Timer(int timeout)
    {
      setTimeout(timeout);
    }

    void setTimeout(int timeout)
    {
      m_timeout = timeout;
    }

    /**
     * @return true if the timer has expired
     */
    boolean tick()
    {
      m_timeout--;
      return m_timeout == 0;
    }

  }

  static class ExpectedAck extends Timer
  {
    public Address sink;
    final Address destination;
    final long sequence;

    ExpectedAck(long aSequence, Address destination, Address sink)
    {
      super(4);
      this.sequence = aSequence;
      this.destination = destination;
      this.sink = sink;
    }

    @Override
    public String toString()
    {
      return "Expect Ack for #" + sequence + " from " + destination;
    }
  }
}
